feat: 增强工具调用参数处理机制 #3036
Open
feat: 增强工具调用参数处理机制 #3036
+31
−2
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
在工具调用时添加参数过滤功能，只传递函数实际需要的参数 解决问题：AstrBotDevs#2988
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
大家好 - 我已审阅了您的更改 - 以下是一些反馈：
- 考虑为每个工具缓存 inspect.signature(handler)，以避免每次调用时重复的自省开销。
- 如果处理程序签名包含 **kwargs，您可以跳过过滤逻辑，以允许灵活的参数，而不是为每个额外的键记录警告。
- 将未知参数汇总到一个警告消息中（例如，列出所有被忽略的键），以减少日志噪音。
给 AI 代理的提示
请处理此代码审查中的评论：
## 总体评论
- 考虑为每个工具缓存 inspect.signature(handler)，以避免每次调用时重复的自省开销。
- 如果处理程序签名包含 **kwargs，您可以跳过过滤逻辑，以允许灵活的参数，而不是为每个额外的键记录警告。
- 将未知参数汇总到一个警告消息中（例如，列出所有被忽略的键），以减少日志噪音。
## 单独评论
### 评论 1
<location> `astrbot/core/agent/runners/tool_loop_agent_runner.py:213` </location>
<code_context>
continue
+ # 参数过滤：只传递函数实际需要的参数
+ import inspect
+ valid_params = {}
+
</code_context>
<issue_to_address>
**issue (complexity):** 考虑在注册时缓存处理程序参数名称或使用 **kwargs，以避免运行时自省并简化参数过滤。
```suggestion
Instead of doing runtime `inspect.signature` in every call, you can cache each handler’s expected params once (e.g. at registration time) and then do a simple dict‐filter. This removes the inline import and loop:
# During tool registration
import inspect
# after setting func_tool.handler
func_tool.param_names = tuple(inspect.signature(func_tool.handler).parameters)
# In _handle_function_tools, replace the entire inspect block with:
valid_params = {
k: v
for k, v in func_tool_args.items()
if k in getattr(func_tool, "param_names", ())
}
# Optional warning for unknown params
for k in func_tool_args:
if k not in valid_params:
logger.warning(f"工具 {func_tool_name} 忽略未知参数: {k}")
await self.agent_hooks.on_tool_start(
self.run_context, func_tool, valid_params
)
executor = self.tool_executor.execute(
tool=func_tool,
run_context=self.run_context,
**valid_params,
)
```
或者更简单，让您的处理程序声明 `**kwargs`，这样您就可以直接传递 `**func_tool_args`，并让 Python 删除未知参数：
```python
# handler example
async def my_tool(required_arg, **kwargs):
...
# call site
executor = self.tool_executor.execute(
tool=func_tool,
run_context=self.run_context,
**func_tool_args,
)
```
这保留了所有行为，消除了运行时自省并减少了样板文件。
</issue_to_address>
### 评论 2
<location> `astrbot/core/agent/runners/tool_loop_agent_runner.py:215-218` </location>
<code_context>
async def _handle_function_tools(
self,
req: ProviderRequest,
llm_response: LLMResponse,
) -> T.AsyncGenerator[MessageChain | list[ToolCallMessageSegment], None]:
"""处理函数工具调用。"""
tool_call_result_blocks: list[ToolCallMessageSegment] = []
logger.info(f"Agent 使用工具: {llm_response.tools_call_name}")
# 执行函数调用
for func_tool_name, func_tool_args, func_tool_id in zip(
llm_response.tools_call_name,
llm_response.tools_call_args,
llm_response.tools_call_ids,
):
try:
if not req.func_tool:
return
func_tool = req.func_tool.get_func(func_tool_name)
logger.info(f"使用工具：{func_tool_name}，参数：{func_tool_args}")
if not func_tool:
logger.warning(f"未找到指定的工具: {func_tool_name}，将跳过。")
tool_call_result_blocks.append(
ToolCallMessageSegment(
role="tool",
tool_call_id=func_tool_id,
content=f"error: 未找到工具 {func_tool_name}",
)
)
continue
# 参数过滤：只传递函数实际需要的参数
import inspect
valid_params = {}
# 获取实际的 handler 函数
handler_func = func_tool.handler
if handler_func:
sig = inspect.signature(handler_func)
for param_name, param_value in func_tool_args.items():
if param_name in sig.parameters:
valid_params[param_name] = param_value
else:
logger.warning(f"工具 {func_tool_name} 忽略未知参数: {param_name}")
else:
# 如果没有 handler（如 MCP 工具），使用所有参数
valid_params = func_tool_args
logger.warning(f"工具 {func_tool_name} 没有 handler，使用所有参数")
try:
await self.agent_hooks.on_tool_start(
self.run_context, func_tool, valid_params
)
except Exception as e:
logger.error(f"Error in on_tool_start hook: {e}", exc_info=True)
executor = self.tool_executor.execute(
tool=func_tool,
run_context=self.run_context,
**valid_params,
)
_final_resp: CallToolResult | None = None
async for resp in executor: # type: ignore
if isinstance(resp, CallToolResult):
res = resp
_final_resp = resp
if isinstance(res.content[0], TextContent):
tool_call_result_blocks.append(
ToolCallMessageSegment(
role="tool",
tool_call_id=func_tool_id,
content=res.content[0].text,
)
)
yield MessageChain().message(res.content[0].text)
elif isinstance(res.content[0], ImageContent):
tool_call_result_blocks.append(
ToolCallMessageSegment(
role="tool",
tool_call_id=func_tool_id,
content="返回了图片(已直接发送给用户)",
)
)
yield MessageChain(type="tool_direct_result").base64_image(
res.content[0].data
)
elif isinstance(res.content[0], EmbeddedResource):
resource = res.content[0].resource
if isinstance(resource, TextResourceContents):
tool_call_result_blocks.append(
ToolCallMessageSegment(
role="tool",
tool_call_id=func_tool_id,
content=resource.text,
)
)
yield MessageChain().message(resource.text)
elif (
isinstance(resource, BlobResourceContents)
and resource.mimeType
and resource.mimeType.startswith("image/")
):
tool_call_result_blocks.append(
ToolCallMessageSegment(
role="tool",
tool_call_id=func_tool_id,
content="返回了图片(已直接发送给用户)",
)
)
yield MessageChain(
type="tool_direct_result"
).base64_image(resource.blob)
else:
tool_call_result_blocks.append(
ToolCallMessageSegment(
role="tool",
tool_call_id=func_tool_id,
content="返回的数据类型不受支持",
)
)
yield MessageChain().message("返回的数据类型不受支持。")
elif resp is None:
# Tool 直接请求发送消息给用户
# 这里我们将直接结束 Agent Loop。
self._transition_state(AgentState.DONE)
if res := self.run_context.event.get_result():
if res.chain:
yield MessageChain(
chain=res.chain, type="tool_direct_result"
)
else:
# 不应该出现其他类型
logger.warning(
f"Tool 返回了不支持的类型: {type(resp)}，将忽略。"
)
try:
await self.agent_hooks.on_tool_end(
self.run_context, func_tool, func_tool_args, _final_resp
)
except Exception as e:
logger.error(f"Error in on_tool_end hook: {e}", exc_info=True)
self.run_context.event.clear_result()
except Exception as e:
logger.warning(traceback.format_exc())
tool_call_result_blocks.append(
ToolCallMessageSegment(
role="tool",
tool_call_id=func_tool_id,
content=f"error: {str(e)}",
)
)
# 处理函数调用响应
if tool_call_result_blocks:
yield tool_call_result_blocks
</code_context>
<issue_to_address>
**suggestion (code-quality):** 我们发现了这些问题：
- 使用命名表达式简化赋值和条件 ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
- 在 ToolLoopAgentRunner.\_handle\_function\_tools 中发现低代码质量 - 10% ([`low-code-quality`](https://docs.sourcery.ai/Reference/Default-Rules/comments/low-code-quality/))
```suggestion
if handler_func := func_tool.handler:
```
<br/><details><summary>解释</summary>
此函数的质量分数低于 25% 的质量阈值。
此分数是方法长度、认知复杂度和工作内存的组合。
如何解决这个问题？
重构此函数以使其更短、更易读可能是有益的。
- 通过将部分功能提取到自己的函数中来减少函数长度。这是您可以做的最重要的事情 - 理想情况下，一个函数应该少于 10 行。
- 减少嵌套，也许通过引入守卫子句来提前返回。
- 确保变量的作用域紧密，以便使用相关概念的代码在函数中坐在一起，而不是分散。
</details>
</issue_to_address>帮助我更有用！请点击 👍 或 👎 对每条评论进行反馈，我将利用这些反馈改进您的评论。
Original comment in English
Hey there - I've reviewed your changes - here's some feedback:
- Consider caching inspect.signature(handler) per tool to avoid repeated introspection overhead on each call.
- If the handler signature includes **kwargs, you can skip the filtering logic to allow flexible parameters rather than logging warnings for each extra key.
- Aggregate unknown parameters into a single warning message (e.g., list all ignored keys) to reduce log noise.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- Consider caching inspect.signature(handler) per tool to avoid repeated introspection overhead on each call.
- If the handler signature includes **kwargs, you can skip the filtering logic to allow flexible parameters rather than logging warnings for each extra key.
- Aggregate unknown parameters into a single warning message (e.g., list all ignored keys) to reduce log noise.
## Individual Comments
### Comment 1
<location> `astrbot/core/agent/runners/tool_loop_agent_runner.py:213` </location>
<code_context>
continue
+ # 参数过滤：只传递函数实际需要的参数
+ import inspect
+ valid_params = {}
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider caching handler parameter names at registration or using **kwargs to avoid runtime introspection and simplify parameter filtering.
```suggestion
Instead of doing runtime `inspect.signature` in every call, you can cache each handler’s expected params once (e.g. at registration time) and then do a simple dict‐filter. This removes the inline import and loop:
# During tool registration
import inspect
# after setting func_tool.handler
func_tool.param_names = tuple(inspect.signature(func_tool.handler).parameters)
# In _handle_function_tools, replace the entire inspect block with:
valid_params = {
k: v
for k, v in func_tool_args.items()
if k in getattr(func_tool, "param_names", ())
}
# Optional warning for unknown params
for k in func_tool_args:
if k not in valid_params:
logger.warning(f"工具 {func_tool_name} 忽略未知参数: {k}")
await self.agent_hooks.on_tool_start(
self.run_context, func_tool, valid_params
)
executor = self.tool_executor.execute(
tool=func_tool,
run_context=self.run_context,
**valid_params,
)
```
Or even simpler, have your handlers declare `**kwargs` so you can pass `**func_tool_args` directly and let Python drop unknown args:
```python
# handler example
async def my_tool(required_arg, **kwargs):
...
# call site
executor = self.tool_executor.execute(
tool=func_tool,
run_context=self.run_context,
**func_tool_args,
)
```
This preserves all behavior, removes runtime introspection and cuts down on boilerplate.
</issue_to_address>
### Comment 2
<location> `astrbot/core/agent/runners/tool_loop_agent_runner.py:215-218` </location>
<code_context>
async def _handle_function_tools(
self,
req: ProviderRequest,
llm_response: LLMResponse,
) -> T.AsyncGenerator[MessageChain | list[ToolCallMessageSegment], None]:
"""处理函数工具调用。"""
tool_call_result_blocks: list[ToolCallMessageSegment] = []
logger.info(f"Agent 使用工具: {llm_response.tools_call_name}")
# 执行函数调用
for func_tool_name, func_tool_args, func_tool_id in zip(
llm_response.tools_call_name,
llm_response.tools_call_args,
llm_response.tools_call_ids,
):
try:
if not req.func_tool:
return
func_tool = req.func_tool.get_func(func_tool_name)
logger.info(f"使用工具：{func_tool_name}，参数：{func_tool_args}")
if not func_tool:
logger.warning(f"未找到指定的工具: {func_tool_name}，将跳过。")
tool_call_result_blocks.append(
ToolCallMessageSegment(
role="tool",
tool_call_id=func_tool_id,
content=f"error: 未找到工具 {func_tool_name}",
)
)
continue
# 参数过滤：只传递函数实际需要的参数
import inspect
valid_params = {}
# 获取实际的 handler 函数
handler_func = func_tool.handler
if handler_func:
sig = inspect.signature(handler_func)
for param_name, param_value in func_tool_args.items():
if param_name in sig.parameters:
valid_params[param_name] = param_value
else:
logger.warning(f"工具 {func_tool_name} 忽略未知参数: {param_name}")
else:
# 如果没有 handler（如 MCP 工具），使用所有参数
valid_params = func_tool_args
logger.warning(f"工具 {func_tool_name} 没有 handler，使用所有参数")
try:
await self.agent_hooks.on_tool_start(
self.run_context, func_tool, valid_params
)
except Exception as e:
logger.error(f"Error in on_tool_start hook: {e}", exc_info=True)
executor = self.tool_executor.execute(
tool=func_tool,
run_context=self.run_context,
**valid_params,
)
_final_resp: CallToolResult | None = None
async for resp in executor: # type: ignore
if isinstance(resp, CallToolResult):
res = resp
_final_resp = resp
if isinstance(res.content[0], TextContent):
tool_call_result_blocks.append(
ToolCallMessageSegment(
role="tool",
tool_call_id=func_tool_id,
content=res.content[0].text,
)
)
yield MessageChain().message(res.content[0].text)
elif isinstance(res.content[0], ImageContent):
tool_call_result_blocks.append(
ToolCallMessageSegment(
role="tool",
tool_call_id=func_tool_id,
content="返回了图片(已直接发送给用户)",
)
)
yield MessageChain(type="tool_direct_result").base64_image(
res.content[0].data
)
elif isinstance(res.content[0], EmbeddedResource):
resource = res.content[0].resource
if isinstance(resource, TextResourceContents):
tool_call_result_blocks.append(
ToolCallMessageSegment(
role="tool",
tool_call_id=func_tool_id,
content=resource.text,
)
)
yield MessageChain().message(resource.text)
elif (
isinstance(resource, BlobResourceContents)
and resource.mimeType
and resource.mimeType.startswith("image/")
):
tool_call_result_blocks.append(
ToolCallMessageSegment(
role="tool",
tool_call_id=func_tool_id,
content="返回了图片(已直接发送给用户)",
)
)
yield MessageChain(
type="tool_direct_result"
).base64_image(resource.blob)
else:
tool_call_result_blocks.append(
ToolCallMessageSegment(
role="tool",
tool_call_id=func_tool_id,
content="返回的数据类型不受支持",
)
)
yield MessageChain().message("返回的数据类型不受支持。")
elif resp is None:
# Tool 直接请求发送消息给用户
# 这里我们将直接结束 Agent Loop。
self._transition_state(AgentState.DONE)
if res := self.run_context.event.get_result():
if res.chain:
yield MessageChain(
chain=res.chain, type="tool_direct_result"
)
else:
# 不应该出现其他类型
logger.warning(
f"Tool 返回了不支持的类型: {type(resp)}，将忽略。"
)
try:
await self.agent_hooks.on_tool_end(
self.run_context, func_tool, func_tool_args, _final_resp
)
except Exception as e:
logger.error(f"Error in on_tool_end hook: {e}", exc_info=True)
self.run_context.event.clear_result()
except Exception as e:
logger.warning(traceback.format_exc())
tool_call_result_blocks.append(
ToolCallMessageSegment(
role="tool",
tool_call_id=func_tool_id,
content=f"error: {str(e)}",
)
)
# 处理函数调用响应
if tool_call_result_blocks:
yield tool_call_result_blocks
</code_context>
<issue_to_address>
**suggestion (code-quality):** We've found these issues:
- Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
- Low code quality found in ToolLoopAgentRunner.\_handle\_function\_tools - 10% ([`low-code-quality`](https://docs.sourcery.ai/Reference/Default-Rules/comments/low-code-quality/))
```suggestion
if handler_func := func_tool.handler:
```
<br/><details><summary>Explanation</summary>
The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.
How can you solve this?
It might be worth refactoring this function to make it shorter and more readable.
- Reduce the function length by extracting pieces of functionality out into
their own functions. This is the most important thing you can do - ideally a
function should be less than 10 lines.
- Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
sits together within the function rather than being scattered.</details>
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
215
to
218
|
|
||
| # 获取实际的 handler 函数 | ||
| handler_func = func_tool.handler | ||
| if handler_func: |
There was a problem hiding this comment.
suggestion (code-quality): 我们发现了这些问题：
- 使用命名表达式简化赋值和条件 (
use-named-expression) - 在 ToolLoopAgentRunner._handle_function_tools 中发现低代码质量 - 10% (
low-code-quality)
Suggested change
| # 获取实际的 handler 函数 | |
| handler_func = func_tool.handler | |
| if handler_func: | |
| if handler_func := func_tool.handler: |
解释
此函数的质量分数低于 25% 的质量阈值。
此分数是方法长度、认知复杂度和工作内存的组合。
如何解决这个问题？
重构此函数以使其更短、更易读可能是有益的。
- 通过将部分功能提取到自己的函数中来减少函数长度。这是您可以做的最重要的事情 - 理想情况下，一个函数应该少于 10 行。
- 减少嵌套，也许通过引入守卫子句来提前返回。
- 确保变量的作用域紧密，以便使用相关概念的代码在函数中坐在一起，而不是分散。
Original comment in English
suggestion (code-quality): We've found these issues:
- Use named expression to simplify assignment and conditional (
use-named-expression) - Low code quality found in ToolLoopAgentRunner._handle_function_tools - 10% (
low-code-quality)
Suggested change
| # 获取实际的 handler 函数 | |
| handler_func = func_tool.handler | |
| if handler_func: | |
| if handler_func := func_tool.handler: |
Explanation
The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.
How can you solve this?
It might be worth refactoring this function to make it shorter and more readable.
- Reduce the function length by extracting pieces of functionality out into
their own functions. This is the most important thing you can do - ideally a
function should be less than 10 lines. - Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
sits together within the function rather than being scattered.
|
过于复杂，而且这并不是从根源上修复。建议不要使用inspect库。 |
不使用`inspect`库，利用现有工具定义信息处理非期望的参数
已经弃用该库，使用 |
请问执行结果正常返回了吗 |
正常的，py执行器不需要参数，有用户输入二次请求执行 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
fixes #2988
Motivation / 动机
解决工具调用时参数不匹配的问题。当 LLM 生成工具调用请求包含额外参数时，会导致
TypeError: got an unexpected keyword argument错误，影响工具正常执行。Modifications / 改动点
tool_loop_agent_runner.py的_handle_function_tools方法中添加参数过滤逻辑tool.parameters["properties"]获取工具函数的参数签名Verification Steps / 验证步骤
python_interpreter工具Screenshots or Test Results / 运行截图或测试结果
Compatibility & Breaking Changes / 兼容性与破坏性变更
Checklist / 检查清单
requirements.txt和pyproject.toml文件相应位置。/ I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations inrequirements.txtandpyproject.toml.Sourcery 总结
增强工具调用，根据函数签名过滤掉意外参数并防止 TypeError，同时记录被忽略的参数并处理没有显式处理程序的工具。
新功能：
错误修复：
改进：
Original summary in English
Summary by Sourcery
Enhance tool invocation to filter out unexpected arguments based on function signatures and prevent TypeErrors, while logging ignored parameters and handling tools without explicit handlers.
New Features:
Bug Fixes:
Enhancements: